package nr.nodehost.DAMiddleLayer;

//import lombok.extern.slf4j.Slf4j;
import nr.nodehost.CmdFrame;
import nr.nodehost.DataFrame;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;

//

//命令行运行带参数：
// java dataAccessComp
// 先启动dataAccess, 然后启动nodeHost.
//@Slf4j
public class dataAccessComp {

    private static Socket cmdSocket8099, dataSocket9999;  // CmdFrame via 8099,  dataFrame via 9999
    private static ConcurrentHashMap<Integer, Integer> schemasCanFinish = new ConcurrentHashMap<>();  // <schemaId, run_stop> , 每一个job, 0表示继续运行, 1表示停止吧
    private static HashMap<Integer, ServerSocket> listenSockets = new HashMap<>();   // dataAccess Listen the port, when stop , this listened port should be free. <schemaId, serverSocket,>

    public static String dataAccessId = "dataAccess";
    public static String dataAccessPort = "dataAccessPort";

    public static void main(String[] args) throws IOException, InterruptedException {

        new Thread(dataAccessComp::listen8099_andAccept).start();

        new Thread(dataAccessComp::connectToNodeHost9999_andSelfIntroduct).start();

        while (cmdSocket8099 == null || dataSocket9999 == null) {
            Thread.sleep(1000);
        }
        // 接收nodeHost给的CmdFrame, 监听新任务的port, 对每一个新任务，启动一个线程处理数据接收，组包，投送给nodehost。
        rev_ParseCmdFrame();
    }

    // Be a  Server, listen 8099, that port is used to connect nodehost
    private static void listen8099_andAccept() {
        try (ServerSocket serverSocket = new ServerSocket(8099)) {
            // nodeHost connect to 8099, we accept and remmember the socket.
            cmdSocket8099 = serverSocket.accept();
            System.out.println("NodeHost connected Port 8099, Get nodeHostClientSocket： " + cmdSocket8099);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // Be a Client, connect To NodeHost via 9999, if connected, make a Self Introduction to Nodehost.
    private static void connectToNodeHost9999_andSelfIntroduct() {
        try {
            while (dataSocket9999 == null) {
                dataSocket9999 = new Socket("127.0.0.1", 9999);
                Thread.sleep(3000);
            }
            System.out.println(" dataAccess connect to nodeHost 9999,  accepted(dataAccessClientSoc): " + dataSocket9999);
            WhoIAm();
            System.out.println(" dataAccess told nodeHost WhoIam ");
        } catch (IOException e) {
            e.printStackTrace();
            connectToNodeHost9999_andSelfIntroduct();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // client to nodehost and tell nodehost i am theDataAccessComponent.
    private static void WhoIAm() throws IOException {
        DataFrame df = new DataFrame();
        df.setSourceComponentId(dataAccessId);
        df.setSourceComponentPort(dataAccessPort);
        df.setTargetComponentId("default");
        df.setTargetComponentPort("default");
        byte[] serializedFrame = df.serialize();
        OutputStream outputStream = dataSocket9999.getOutputStream();
        outputStream.write(serializedFrame);
        outputStream.flush();
    }

    // 接收命令帧，解析并执行相应的处理逻辑。
    private static void rev_ParseCmdFrame() {
        do {
            CmdFrame receivedFrame = CmdFrame.readOneFrame(cmdSocket8099);
            System.out.println("dataAccess rev_cmdframe: " + receivedFrame);
            int dataAccess_startStop = receivedFrame.getDataAccess_StartStop();  // 0 start, 1 stop
            int schemaId = receivedFrame.getSchemaId();

            if (dataAccess_startStop == 0) { // start
                startNewDataFlow(schemaId, receivedFrame.getPort_dataRemote(), receivedFrame.getSrcFrameLength());
            } else {  // stop
                stopDataFlow(schemaId);
            }
        } while (true);
    }

    // 用一个线程负责一个任务流。
    private static void startNewDataFlow(int schemaId, int port, int frameLength) {
        schemasCanFinish.put(schemaId, 0);  // 0 is start
        new Thread(() -> {
            newJob(schemaId, port, frameLength);
        }).start();
    }

    // As a server, 监听port，并建立与数据源通信的 socket， 启动接收-封包-投递给nodeHost任务,
    public static void newJob(int schemaId, int port, int srcframeLength) {
        try (ServerSocket serverSocket = new ServerSocket(port)) {
            listenSockets.put(schemaId, serverSocket);
            // dataSrc process  connect to the port, we accept and bind the socket to the job
            Socket socket = serverSocket.accept();
            //如果建立了数据源和dataAccess的数据连接，那么socket可以去做通信，但不接受新的客户端连接请求， 也移除listenSockets的相应元素。
            serverSocket.close();
            listenSockets.remove(schemaId);
            System.out.println("dataSrc connected Port: " + port);  // 每一个任务一个port, 一个socket, 由于对应一个线程负责，没必要保存sockets列表。
            //tell webserver via http that the port is  connected by dataSrc, the port is used.
            InputStream is = socket.getInputStream();
            byte[] buf = new byte[srcframeLength];
            if (is != null && buf != null) {
                //  0 表示正常运行， 1 表示关闭
                DataFrame df = new DataFrame();
                while (schemasCanFinish.get(schemaId).equals(0)) {
                    // read socket. inputstream
                    int nread = DataFrame.sRead(is, buf);
                    // make dataframe
                    df.setSchemaId(schemaId);
                    df.setSourceComponentId(dataAccessId);
                    df.setSourceComponentPort(dataAccessPort);
                    df.setTargetComponentId("default");
                    df.setTargetComponentPort("default");
                    df.setLength(srcframeLength);
                    df.setData(buf);
                    // send to nodehost via 9999
                    System.out.println(df);
                    sendDataFrame(dataSocket9999.getOutputStream(), df);
                    System.out.println("dataAccess sendDataFrame to nodeHost " + buf);
                }
            }
            socket.close();  // 这里表示该任务应该结束，关闭socket就好。
            System.out.println("socket closed " + socket + " with port: " + port);
            //   tell webserver via http that the job is stopped, the port is now free again.
        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
    }

    // 关闭该schemaId对应的socket,结束数据接收
    private static void stopDataFlow(int schemaId) {
        schemasCanFinish.replace(schemaId, 1);  // 1 is stop， when 1 is set , datasocket(dataSrc - dataAccess) can be closed.
        System.out.println("set schemaId  stop flag " + schemaId);
        ServerSocket serverSocket = listenSockets.get(schemaId);
        if (serverSocket != null) {
            try {
                serverSocket.close();
                System.out.println("serverSocket is closed of schemaId: " + schemaId);
            } catch (IOException e) {
                System.out.println(e.getMessage());
            }
        }

    }

    private static void sendDataFrame(OutputStream outputStream, DataFrame frame) throws IOException {
        try {
            // Serialize the DataFrame
            byte[] serializedFrame = frame.serialize();
            // Send the serialized DataFrame
            outputStream.write(serializedFrame);
            outputStream.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
